package com.bigdata

import java.sql.Timestamp
import java.util.Properties

import com.bigdata.bean.{ItemViewCount, UserBehavior}
import com.bigdata.common.constants.Constants
import com.bigdata.common.utils.KafkaCommonUtils
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.core.fs.FileSystem.WriteMode
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

object HotItems {
  def main(args: Array[String]): Unit = {

    // 创建一个 StreamExecutionEnvironment
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // 设定Time类型为EventTime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val logDataStream: DataStream[String] = KafkaCommonUtils.createKafkaSink(env, Constants.Topic_Item)

    //    val logDataStream: DataStream[String] = env.readTextFile("E:\\BigData\\UserBehaviorAnalysis\\HotItemsAnalysis\\src\\main\\resources\\UserBehavior.csv")

    val stream: DataStream[String] = logDataStream.map { line =>
      val itemArray: Array[String] = line.split(",")
      UserBehavior(itemArray(0).toLong, itemArray(1).toLong, itemArray(2).toInt, itemArray(3), itemArray(4).toLong)
    }
      .assignAscendingTimestamps(_.timestamp * 1000)
      .filter(_.behavior == "pv")
      .keyBy(_.itemId)
      .timeWindow(Time.minutes(60), Time.minutes(5))
      .aggregate(new CountAgg, new WindowResultFunction)
      .keyBy(_.windowEnd)
      .process(new TopNHotItems(3))

    stream.print().setParallelism(1)
    stream.writeAsText("./output/hotItems.txt", WriteMode.OVERWRITE).setParallelism(1)

    env.execute("Hot Items Count Job")


  }

  class CountAgg extends AggregateFunction[UserBehavior, Long, Long] {
    override def createAccumulator(): Long = 0L

    override def add(in: UserBehavior, acc: Long): Long = acc + 1L

    override def getResult(acc: Long): Long = acc

    override def merge(acc: Long, acc1: Long): Long = acc + acc1
  }

  class WindowResultFunction extends WindowFunction[Long, ItemViewCount, Long, TimeWindow] {
    override def apply(key: Long, window: TimeWindow, input: Iterable[Long], out: Collector[ItemViewCount]): Unit = {
      val count = input.iterator.next()
      out.collect(ItemViewCount(key, window.getEnd, count))
    }
  }

  class TopNHotItems(n: Int) extends KeyedProcessFunction[Long, ItemViewCount, String] {

    lazy val itemState: ListState[ItemViewCount] = getRuntimeContext.getListState(new ListStateDescriptor[ItemViewCount]("itemState-state", classOf[ItemViewCount]))

    override def processElement(i: ItemViewCount, context: KeyedProcessFunction[Long, ItemViewCount, String]#Context, collector: Collector[String]): Unit = {
      // 每条数据都保存到状态中
      itemState.add(i)

      context.timerService.registerEventTimeTimer(i.windowEnd + 1)

    }

    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, ItemViewCount, String]#OnTimerContext, out: Collector[String]): Unit = {
      import scala.collection.JavaConversions._
      val listBuffer: ListBuffer[ItemViewCount] = ListBuffer()
      // 获取收到的所有商品点击量
      for (item <- itemState.get()) (
        listBuffer.+=(item)
        )

      // 提前清除状态中的数据，释放空间
      itemState.clear()

      // 按照点击量从大到小排序
      val sortedItems: ListBuffer[ItemViewCount] = listBuffer.sortBy(_.count)(Ordering.Long.reverse).take(n)

      // 将排名信息格式化成 String, 便于打印
      val result: StringBuilder = new StringBuilder
      result.append("====================================\n")
      result.append(s"时间: ${new Timestamp(timestamp - 1)}\n")

      for (i <- sortedItems.indices) {
        val currentItem: ItemViewCount = sortedItems(i)
        // e.g.  No1：  商品ID=12224  浏览量=2413
        result.append(s"No.${i + 1}:\t商品ID=${currentItem.itemId}\t浏览量=${currentItem.count}\n")

      }
      result.append("====================================\n\n")
      // 控制输出频率，模拟实时滚动结果
      //      Thread.sleep(1000)
      out.collect(result.toString)
    }
  }


}
